packages/core/src/v3/test/test-realtime-streams-manager.ts-158-161 (1)

158-161: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

reset() leaks writeListeners across tests.

reset() clears buffers and pipeWaits but leaves writeListeners intact. For a manager whose primary purpose is per-test isolation, any listener registered in one test (and not explicitly unsubscribed) will continue to fire for writes in subsequent tests that share this instance, leading to flaky/cross-talking tests.

🧹 Proposed fix

   reset(): void {
     this.buffers.clear();
     this.pipeWaits.clear();
+    this.writeListeners.clear();
   }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/test/test-realtime-streams-manager.ts` around lines 158
- 161, The reset() method currently clears buffers and pipeWaits but leaves
writeListeners populated, causing listeners to leak between tests; update the
reset() implementation (the reset() method in the test realtime streams manager)
to also remove all registered write listeners (e.g., clear or reinitialize the
writeListeners collection) so any callbacks registered via writeListeners are
unsubscribed/cleared between tests to ensure per-test isolation.

packages/core/src/v3/apiClient/runStream.ts-480-491 (1)

480-491: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Abort during backoff is not wakeable

The retry delay promise only listens to retryNowController. If caller signal aborts during backoff, shutdown can be delayed until the timer expires.

Suggested fix

     this.retryNowController = new AbortController();
     await new Promise<void>((resolve) => {
+      if (this.options.signal?.aborted) {
+        resolve();
+        return;
+      }
       const timer = setTimeout(() => {
         this.retryNowController?.signal.removeEventListener("abort", onAbort);
+        this.options.signal?.removeEventListener("abort", onUserAbort);
         resolve();
       }, delay);
       const onAbort = () => {
         clearTimeout(timer);
+        this.options.signal?.removeEventListener("abort", onUserAbort);
+        resolve();
+      };
+      const onUserAbort = () => {
+        clearTimeout(timer);
+        this.retryNowController?.signal.removeEventListener("abort", onAbort);
         resolve();
       };
       this.retryNowController!.signal.addEventListener("abort", onAbort, { once: true });
+      this.options.signal?.addEventListener("abort", onUserAbort, { once: true });
     });

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/apiClient/runStream.ts` around lines 480 - 491, The
backoff promise only listens to this.retryNowController and ignores the caller's
abort signal, causing shutdown to wait for the timer; update the wait logic in
runStream.ts to also listen to the external/caller signal (e.g., this.signal or
the passed-in signal) alongside this.retryNowController by adding an event
listener on that signal which clears the timeout, resolves the promise, and
removes both listeners (mirror the existing onAbort cleanup for
retryNowController.signal and the timer) to ensure immediate wake-up on caller
abort while preventing listener leaks.

packages/core/src/v3/inputStreams/manager.ts-181-188 (1)

181-188: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

disconnectStream may be silently undone by auto-reconnect when handlers or waiters still exist.

The tail's .finally block auto-reconnects whenever this.handlers or this.onceWaiters still have entries for streamId. If any .on() handler or pending .once() waiter is active when disconnectStream runs, the abort will complete but the tail immediately reconnects—negating the contract: "disconnect before .wait() suspends so the tail doesn't buffer duplicates delivered through the waitpoint path."

Add an explicitlyDisconnected flag (as already implemented in sessionStreams/manager.ts) to prevent auto-reconnect after intentional disconnect. Mark the stream before abort, check the flag in the .finally reconnect branch, and clear it on the next on()/once() call.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/inputStreams/manager.ts` around lines 181 - 188, The
disconnectStream implementation currently aborts and removes the tail but can be
immediately undone by the tail's .finally auto-reconnect if handlers or
onceWaiters still reference streamId; add an explicitlyDisconnected flag (same
pattern used in sessionStreams/manager.ts) to the manager: set
explicitlyDisconnected[streamId] = true before calling
tail.abortController.abort() in disconnectStream, check explicitlyDisconnected
inside the tail's .finally reconnect branch to skip reconnect when true, and
ensure explicitlyDisconnected[streamId] is cleared when a new .on() or .once()
call for that streamId occurs so reconnects resume normally.

packages/core/src/v3/test/test-session-stream-manager.ts-215-247 (1)

215-247: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Handler invocation in __sendFromTest doesn't use the safe invoke helper — and the helper itself is dead.

__sendFromTest calls handlers directly via Array.from(handlers).map((h) => Promise.resolve().then(() => h(data))) (Line 225). A thrown/rejected handler will reject the Promise.all and propagate out of __sendFromTest, which is the opposite of the comment on Line 278 ("Never let a handler error break test state"). Meanwhile, the invoke private method defined on Lines 270-279 (which does exactly that catch) is never referenced anywhere in the file — dead code.

Either drop invoke, or route handler dispatch through it so a misbehaving handler doesn't blow up the test harness.

🛡️ Suggested fix

     const handlers = this.handlers.get(key);
     if (handlers && handlers.size > 0) {
-      await Promise.all(
-        Array.from(handlers).map((h) => Promise.resolve().then(() => h(data)))
-      );
+      await Promise.all(
+        Array.from(handlers).map(async (h) => {
+          try {
+            await h(data);
+          } catch {
+            // Never let a handler error break test state
+          }
+        })
+      );
     }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/test/test-session-stream-manager.ts` around lines 215 -
247, The handler loop in __sendFromTest currently calls handlers directly with
Promise.all(Array.from(handlers).map((h) => Promise.resolve().then(() =>
h(data)))) so a thrown/rejected handler will reject the whole send; instead
reuse the existing private invoke helper (or restore its implementation) to
swallow or handle handler errors as intended. Replace the direct handler
invocation in __sendFromTest to call this.invoke(h, data) for each handler
(i.e., Promise.resolve().then(() => this.invoke(h, data)) or map to this.invoke)
so individual handler failures are caught and do not break test state; if invoke
is currently dead/incorrect, fix its implementation to catch errors (and
optionally log) and return a resolved promise. Ensure the change references the
handlers map and the invoke method so behavior is consistent with the "Never let
a handler error break test state" comment.

packages/core/src/v3/sessionStreams/manager.ts-121-145 (1)

121-145: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Abort listener and timeout handle leak on normal once() resolution.

When a once() waiter resolves via #dispatch (Line 359), neither the signal's abort listener nor the unused timeout handle are removed/cleared. The abort listener is registered with { once: true }, so it self-clears if the signal eventually fires, but in the common case where the signal outlives this once() (e.g., a long-lived run-level AbortSignal reused across many once() calls), the listener — which captures the waiter and this — accumulates on the signal until it finally aborts. #dispatch should also clear the timeout handle (it currently only clears it on Line 358 for the leading waiter via clearTimeout) — wait, it does — but it never invokes signal.removeEventListener.

Compare with TestSessionStreamManager which keeps abortHandler on the waiter and removes it on dispatch (packages/core/src/v3/test/test-session-stream-manager.ts, Lines 234-236).

🛡️ Suggested fix

 type OnceWaiter = {
   resolve: (result: InputStreamOnceResult<unknown>) => void;
   reject: (error: Error) => void;
   timeoutHandle?: ReturnType<typeof setTimeout>;
+  signal?: AbortSignal;
+  abortHandler?: () => void;
 };

     return new InputStreamOncePromise<unknown>((resolve, reject) => {
       const waiter: OnceWaiter = { resolve, reject };

       if (options?.signal) {
         if (options.signal.aborted) {
           reject(new Error("Aborted"));
           return;
         }
-        options.signal.addEventListener(
-          "abort",
-          () => {
-            if (waiter.timeoutHandle) clearTimeout(waiter.timeoutHandle);
-            this.#removeOnceWaiter(key, waiter);
-            reject(new Error("Aborted"));
-          },
-          { once: true }
-        );
+        const abortHandler = () => {
+          if (waiter.timeoutHandle) clearTimeout(waiter.timeoutHandle);
+          this.#removeOnceWaiter(key, waiter);
+          reject(new Error("Aborted"));
+        };
+        waiter.signal = options.signal;
+        waiter.abortHandler = abortHandler;
+        options.signal.addEventListener("abort", abortHandler, { once: true });
       }

And in #dispatch (and the buffered-shift path) clear the listener when the waiter is resolved:

       const waiter = waiters.shift()!;
       if (waiters.length === 0) this.onceWaiters.delete(key);
       if (waiter.timeoutHandle) clearTimeout(waiter.timeoutHandle);
+      if (waiter.signal && waiter.abortHandler) {
+        waiter.signal.removeEventListener("abort", waiter.abortHandler);
+      }
       waiter.resolve({ ok: true, output: data });

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/sessionStreams/manager.ts` around lines 121 - 145, The
once() waiter installs an abort listener and timeout but does not remove the
abort listener when the waiter is resolved via `#dispatch`, causing
listener/closure leaks; update once() to store the abort handler on the waiter
(e.g., waiter.abortHandler) when adding options.signal.addEventListener, and in
`#dispatch` (and any buffered-shift resolution path) call
options.signal.removeEventListener("abort", waiter.abortHandler) and
clearTimeout(waiter.timeoutHandle) when resolving the waiter so both the
listener and timeout are cleaned up; reference the waiter object used in once()
and the `#dispatch` method to apply these removals.

apps/webapp/app/services/runsRepository/runsRepository.server.ts-45-46 (1)

45-46: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Constrain taskKinds to known values.

Line 45 currently accepts any string, so invalid values can silently flow into filtering and produce confusing empty results. Tightening this to known task kinds improves input correctness.

Suggested diff

 const RunStatus = z.enum(Object.values(TaskRunStatus) as [TaskRunStatus, ...TaskRunStatus[]]);
+const TaskKind = z.enum(["TASK", "AGENT", "SCHEDULED"]);

 const RunListInputOptionsSchema = z.object({
@@
-  taskKinds: z.array(z.string()).optional(),
+  taskKinds: z.array(TaskKind).optional(),
 });

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/webapp/app/services/runsRepository/runsRepository.server.ts` around
lines 45 - 46, The taskKinds schema currently allows any string (taskKinds:
z.array(z.string()).optional()), which lets invalid values pass; change it to
constrain to the known task kinds (e.g., replace z.string() with z.enum([...])
or z.nativeEnum(TaskKind) referencing your central TaskKind enum) so only valid
task kind values are accepted; update imports to bring in the TaskKind enum or
explicitly list allowed strings and adjust any callers/types if needed.

apps/webapp/app/routes/resources.sessions.$sessionParam.close.ts-10-14 (1)

10-14: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Reuse the shared close-session schema here.

This local schema has already drifted from the public contract: it accepts arbitrarily long reason strings, while CloseSessionRequestBody caps them at 256 chars. Import the shared schema (or at least mirror its limit) so the dashboard and API reject the same payloads.

As per coding guidelines: apps/webapp/**/*.{ts,tsx} should use subpath exports from @trigger.dev/core package instead of importing from the root @trigger.dev/core path.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/webapp/app/routes/resources.sessions`.$sessionParam.close.ts around
lines 10 - 14, The local closeSessionSchema has drifted from the shared
CloseSessionRequestBody contract (it allows unlimited reason length); update
closeSessionSchema to match the shared schema by either importing the shared
CloseSessionRequestBody schema from the `@trigger.dev/core` subpath export and
using it directly, or at minimum constrain reason to z.string().max(256). Ensure
the import uses a subpath export from `@trigger.dev/core` (e.g. import {
CloseSessionRequestBody } from '@trigger.dev/core/…') rather than importing from
the package root, and replace or remove the local schema definition accordingly.

apps/webapp/app/routes/realtime.v1.sessions.$session.$io.records.ts-18-23 (1)

18-23: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Reject oversized cursors before coercing them to Number.

afterEventId currently accepts any digit string via z.string().regex(/^\d+$/), but line 74 coerces it unsafely with Number(searchParams.afterEventId). Long inputs can become Infinity or lose precision, causing the endpoint to skip/duplicate records or pass an invalid cursor downstream. Tighten SearchSchema to validate finite safe integers before conversion.

Suggested fix

 const SearchSchema = z.object({
   // S2 sequence number — same cursor format as the SSE Last-Event-ID
   // (the SSE `id:` field on session-channel events is the seq_num,
   // stringified). Records returned have `seqNum > afterEventId`.
-  afterEventId: z.string().regex(/^\d+$/).optional(),
+  afterEventId: z
+    .string()
+    .regex(/^\d+$/)
+    .refine((value) => Number.isSafeInteger(Number(value)), {
+      message: "afterEventId must be a safe integer",
+    })
+    .optional(),
 });

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/webapp/app/routes/realtime.v1.sessions`.$session.$io.records.ts around
lines 18 - 23, SearchSchema currently allows any digit string for afterEventId
but the code later does Number(searchParams.afterEventId) (see usage of Number
in the handler), which can produce Infinity or lose precision for oversized
values; update SearchSchema to reject values outside JavaScript safe integer
range (or use a z.preprocess that coerces the string to a number and validates
Number.isFinite() && Number.isSafeInteger() and >= 0) so only finite safe
non-negative integers are accepted before conversion, then replace the unsafe
Number(...) usage with the validated numeric value from the parsed schema (keep
references to SearchSchema and afterEventId and the Number(...) conversion to
locate the change).

apps/webapp/app/components/runs/v3/TaskRunsTable.tsx-357-361 (1)

357-361: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Remove the unsafe type cast for taskKind.

The taskKind field is typed as TaskKind, which allows arbitrary strings via .or(anyString) in the Zod schema, while TaskTriggerSource is limited to three specific enum values: STANDARD, SCHEDULED, and AGENT. The cast bypasses type safety and could pass invalid values to TaskTriggerSourceIcon. Either narrow the TaskKind schema to exclude arbitrary strings, or validate the value before casting.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/webapp/app/components/runs/v3/TaskRunsTable.tsx` around lines 357 - 361,
The code unsafely casts run.taskKind to TaskTriggerSource before passing it to
TaskTriggerSourceIcon; instead validate or narrow the value: check that
run.taskKind is one of the TaskTriggerSource enum values (e.g., compare against
Object.values(TaskTriggerSource) or use a helper isTaskTriggerSource) and only
pass it when valid, otherwise pass a safe fallback (or undefined) or render a
default icon; alternatively tighten the TaskKind schema to only allow the
TaskTriggerSource union so no cast is needed. Ensure references:
TaskRunsTable.tsx, run.taskKind, TaskTriggerSource, TaskTriggerSourceIcon, and
TaskKind.

apps/webapp/app/routes/resources.orgs.$organizationSlug.projects.$projectParam.env.$envParam.sessions.$sessionParam.realtime.v1.$io.ts-61-69 (1)

61-69: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Reject zero and partially-numeric Timeout-Seconds values.

Line 63 uses parseInt(), so values like "10foo" are accepted, and Line 66’s truthy guard lets "0" through even though this route documents a 1..600 range.

Suggested fix

-  const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined;
-  const timeoutInSeconds = timeoutInSecondsRaw ? parseInt(timeoutInSecondsRaw) : undefined;
+  const timeoutInSecondsRaw = request.headers.get("Timeout-Seconds") ?? undefined;
+  const timeoutInSeconds =
+    timeoutInSecondsRaw === undefined ? undefined : Number(timeoutInSecondsRaw);

-  if (
-    timeoutInSeconds &&
-    (isNaN(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600)
-  ) {
+  if (
+    timeoutInSeconds !== undefined &&
+    (!Number.isInteger(timeoutInSeconds) || timeoutInSeconds < 1 || timeoutInSeconds > 600)
+  ) {
     return new Response("Invalid timeout", { status: 400 });
   }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In
`@apps/webapp/app/routes/resources.orgs`.$organizationSlug.projects.$projectParam.env.$envParam.sessions.$sessionParam.realtime.v1.$io.ts
around lines 61 - 69, The timeout parsing currently uses parseInt on
timeoutInSecondsRaw (from the "Timeout-Seconds" header) which accepts
partially-numeric strings like "10foo" and treats "0" as falsy; change the
validation in the timeoutInSeconds calculation (and the subsequent if) to first
verify the header matches a pure integer regex (e.g. /^\d+$/) before parsing,
then parse to a number (Number or parseInt) and explicitly reject 0 and values
outside 1..600; update the checks around timeoutInSecondsRaw/timeoutInSeconds
and the error return to ensure partially-numeric and "0" values return 400.

apps/webapp/app/components/runs/v3/RunFilters.tsx-194-196 (1)

194-196: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Validate sources against the supported trigger-source set.

Right now sources accepts any non-empty string, so ?sources=foo survives parsing, shows up as an applied filter, and gets passed downstream even though this UI only supports STANDARD, SCHEDULED, and AGENT.

Suggested fix

+const RunSource = z.enum(["STANDARD", "SCHEDULED", "AGENT"]);
+
 export const TaskRunListSearchFilters = z.object({
   cursor: z.string().optional().describe("Cursor for pagination - used internally for navigation"),
   direction: z
     .enum(["forward", "backward"])
     .optional()
@@
-  sources: StringOrStringArray.describe(
+  sources: z
+    .preprocess((value) => {
+      if (typeof value === "string") {
+        return value.length > 0 ? [value] : undefined;
+      }
+
+      if (Array.isArray(value)) {
+        return value.filter((v) => typeof v === "string" && v.length > 0);
+      }
+
+      return undefined;
+    }, RunSource.array().optional())
+    .describe(
     "Task trigger sources to filter by (STANDARD, SCHEDULED, AGENT)"
-  ),
+  ),
 });

As per coding guidelines, {packages/core,apps/webapp}/**/*.{ts,tsx}: Use zod for validation in packages/core and apps/webapp.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/webapp/app/components/runs/v3/RunFilters.tsx` around lines 194 - 196,
Replace the loose StringOrStringArray.describe for the "sources" field with a
zod-based schema that only allows the supported trigger sources: "STANDARD",
"SCHEDULED", "AGENT". Specifically, change the "sources" schema in
RunFilters.tsx to use zod.enum (or z.union of z.string and z.array(z.enum(...))
if you need single-or-array behavior) instead of StringOrStringArray.describe,
and ensure parsing/validation uses that zod schema so queries like ?sources=foo
are rejected/normalized before becoming an applied filter or passed downstream.

packages/core/src/v3/test/test-realtime-streams-manager.ts (1)

22-22: 💤 Low value

pipeWaits is write-only dead state.

pipeWaits is populated in pipe() (lines 83-84) and cleared in reset() (line 160), but nothing ever reads from it. Either remove it, or expose the waitAll(key?) helper this state seems to anticipate (useful for tests that fan out multiple pipe() calls and want to await them collectively without holding each returned instance).

♻️ Option A — drop the unused field

-  private pipeWaits = new Map<string, Promise<void>[]>();
   private writeListeners = new Set<WriteListener>();

-    if (!this.pipeWaits.has(key)) this.pipeWaits.set(key, []);
-    this.pipeWaits.get(key)!.push(done);
-
     return {

   reset(): void {
     this.buffers.clear();
-    this.pipeWaits.clear();
   }

Option B — keep it and add a __waitAllFromTest(key?) helper that awaits the tracked done promises so tests can drain before assertions.

Also applies to: 83-84, 160-160

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/test/test-realtime-streams-manager.ts` at line 22, The
pipeWaits Map field is never read (populated in pipe() and cleared in reset())
so either remove pipeWaits entirely or implement a test-facing waiter so it’s
useful; to fix, choose one: (A) delete the private pipeWaits property and any
code that pushes into it inside pipe() and reset(), or (B) keep pipeWaits and
add a method like __waitAllFromTest(key?: string) / waitAll(key?: string) that
returns Promise<void> which awaits Promise.all(this.pipeWaits.get(key) || [])
and clears the stored array, and update reset() to use that helper to drain
entries; reference the pipeWaits field, the pipe() method where entries are
added, and the reset() method where they’re cleared when making the change.

packages/core/src/v3/test/test-input-stream-manager.ts (1)

117-121: ⚡ Quick win

Test manager's shiftBuffer/disconnectStream diverge from real manager semantics.

shiftBuffer always returns false and disconnectStream is a complete no-op, but the real StandardInputStreamManager shifts/clears the buffered head used by once(). Since pendingSends plays exactly that role in this test fixture (drained by once() at lines 62–68), tests that exercise .wait()-style flows (drop-the-duplicate / disconnect-before-suspend) will not observe correct behavior here.

Aligning the stubs would make tests faithfully reproduce production semantics.

♻️ Suggested alignment

-  shiftBuffer(_streamId: string): boolean {
-    return false;
-  }
-
-  disconnectStream(_streamId: string): void {}
+  shiftBuffer(streamId: string): boolean {
+    const buffered = this.pendingSends.get(streamId);
+    if (buffered && buffered.length > 0) {
+      buffered.shift();
+      if (buffered.length === 0) this.pendingSends.delete(streamId);
+      return true;
+    }
+    return false;
+  }
+
+  disconnectStream(streamId: string): void {
+    this.pendingSends.delete(streamId);
+  }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/test/test-input-stream-manager.ts` around lines 117 -
121, The test fixture's shiftBuffer and disconnectStream do not mimic production
semantics: update shiftBuffer(streamId) to remove/shift the buffered head from
the pendingSends structure for that stream and return true when it actually
removed an entry (false otherwise), and update disconnectStream(streamId) to
clear any buffered entries in pendingSends for that stream (and/or call
shiftBuffer repeatedly) so that the once() consumer behavior that drains
pendingSends is reproduced; reference the shiftBuffer and disconnectStream
methods and the pendingSends collection and once() consumer when making the
changes.

packages/core/src/v3/sessionStreams/types.ts (1)

24-76: 💤 Low value

Prefer type alias over interface for the new SessionStreamManager declaration.

Per the repository's TypeScript guideline, types are preferred over interfaces. Since this is a brand-new declaration in a new file, it's a good opportunity to follow the guideline (the pre-existing InputStreamManager interface can be migrated separately).

♻️ Proposed change

-export interface SessionStreamManager {
-  /** Register a handler that fires every time data arrives on the given channel. */
-  on(
-    sessionId: string,
-    io: SessionChannelIO,
-    handler: (data: unknown) => void | Promise<void>
-  ): { off: () => void };
-  ...
-}
+export type SessionStreamManager = {
+  /** Register a handler that fires every time data arrives on the given channel. */
+  on(
+    sessionId: string,
+    io: SessionChannelIO,
+    handler: (data: unknown) => void | Promise<void>
+  ): { off: () => void };
+  ...
+};

As per coding guidelines: "Use types over interfaces for TypeScript".

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/sessionStreams/types.ts` around lines 24 - 76, The
declaration for SessionStreamManager should be converted from an interface to a
type alias per repository TypeScript guidelines: replace the "interface
SessionStreamManager { ... }" with "type SessionStreamManager = { ... }"
preserving all member signatures (on, once, peek, lastSeqNum, setLastSeqNum,
setMinTimestamp, shiftBuffer, disconnectStream, clearHandlers, reset,
disconnect) and exported name; no API or runtime behavior should change, just
the syntactic form (leave InputStreamManager untouched for separate migration).

packages/core/src/v3/sessionStreams/manager.ts (1)

137-145: 💤 Low value

timeoutMs: 0 silently disables the timeout.

The truthy check on Line 137 treats timeoutMs: 0 as "no timeout", but TestSessionStreamManager.once (test-session-stream-manager.ts Line 102) uses options?.timeoutMs !== undefined and would arm a 0ms timer that resolves immediately. The two implementations therefore disagree on 0. If 0 is not a meaningful "fire immediately" value here, leave a comment; otherwise switch to an explicit !== undefined check for parity.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/sessionStreams/manager.ts` around lines 137 - 145, The
current truthy check in the once waiter (where waiter.timeoutHandle is set)
treats timeoutMs: 0 as unset and thus disables the timer; align behavior with
TestSessionStreamManager.once by checking explicitly for undefined (use
options.timeoutMs !== undefined) so a 0ms timeout is honored; update the
condition around waiter.timeoutHandle and ensure the timeout callback still uses
options.timeoutMs (and constructs the InputStreamTimeoutError(key,
options.timeoutMs!)) and leaves `#removeOnceWaiter` and resolve logic unchanged.

packages/core/src/v3/test/test-session-stream-manager.ts (2)

144-146: 💤 Low value

setLastSeqNum is unconditionally monotonic in production but not here.

StandardSessionStreamManager.setLastSeqNum only advances the stored seqNum when seqNum > current (packages/core/src/v3/sessionStreams/manager.ts Lines 168-171). The test manager overwrites unconditionally, which can mask a class of regressions where production-side regression is the whole point of the test (e.g., resume-after-reconnect with out-of-order acks). Consider matching production semantics unless there's a specific reason to diverge.

♻️ Suggested fix

   setLastSeqNum(sessionId: string, io: SessionChannelIO, seqNum: number): void {
-    this.seqNums.set(keyFor(sessionId, io), seqNum);
+    const key = keyFor(sessionId, io);
+    const current = this.seqNums.get(key);
+    if (current === undefined || seqNum > current) {
+      this.seqNums.set(key, seqNum);
+    }
   }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/test/test-session-stream-manager.ts` around lines 144 -
146, The test implementation of setLastSeqNum currently overwrites seqNums
unconditionally; change it to match production behavior in
StandardSessionStreamManager by only updating this.seqNums for keyFor(sessionId,
io) when the incoming seqNum is greater than the existing value (i.e., retrieve
current via this.seqNums.get(...), compare and only call this.seqNums.set(...)
if seqNum > current), preserving monotonic advancement semantics used for
resume-after-reconnect/out-of-order ack tests.

---

197-214: 💤 Low value

Docstring misstates production behavior.

The comment on Lines 207-213 says "Production discards records that only match handlers — but in production the SSE tail introduces enough latency that the next .once is usually registered before the next record arrives." Looking at StandardSessionStreamManager.#dispatch (packages/core/src/v3/sessionStreams/manager.ts Lines 364-378), production also buffers the record when there's no waiter (handlers fire and buffer is appended). The test manager's behavior actually mirrors production here; the only deviation is that production buffers from onPart, not from a test-driver entry point. Update the comment so future readers don't get a misleading mental model.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/test/test-session-stream-manager.ts` around lines 197 -
214, The docstring in test-session-stream-manager.ts incorrectly claims
production discards records that only match handlers; update the comment to
reflect that StandardSessionStreamManager.#dispatch also appends to the buffer
when no waiter is present (handlers fire and the record is buffered), and
clarify that the real difference is where buffering is triggered (production
buffers from onPart while the test manager buffers from the test-driver entry
point). Edit the block describing dispatch rules (the paragraph starting with
"If no waiter, the record is buffered...") to state this corrected behavior and
mention the actual deviation between production and tests (buffer origin),
referencing the TestSessionStreamManager and
StandardSessionStreamManager.#dispatch to guide readers.

packages/core/src/v3/sessionStreams/manager.test.ts (2)

101-134: 💤 Low value

Test name vs. assertion is slightly off; consider tightening.

The test claims it verifies that a filter on "in" doesn't bleed into "out", but the first manager only ever asserts on "out" and never proves the "in" filter on that same manager (Line 113 receives { kind: "in-record" }, which the singleShot mock delivers to whichever io subscribes first). The actual filter assertion uses a second manager (manager2) with separate state (Lines 120-127). The proof you want — same manager, same session, different io, only one is filtered — isn't actually exercised. Consider building one mock that routes records per io (e.g., emit different fixtures on different _io calls) so a single manager can demonstrate the per-io isolation end-to-end.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/sessionStreams/manager.test.ts` around lines 101 - 134,
The test currently proves filtering on "in" using a separate manager (manager2)
instead of demonstrating per-(sessionId, io) isolation on the same
StandardSessionStreamManager; update the test to use a single
StandardSessionStreamManager and a single mock API that emits different fixture
records depending on the requested _io so that the same manager subscribes to
both "in" and "out" and you can call setMinTimestamp(sessionId, "in", 5000) then
assert once(sessionId, "out") returns the expected record while once(sessionId,
"in") times out; modify or replace singleShotApiClient to inspect the _io
argument and return distinct records for "in" vs "out", then keep using
manager.disconnectStream/manager.disconnect to clean up.

---

13-44: 💤 Low value

Mock cast bypasses ApiClient type checking.

as unknown as ApiClient (Line 43) silently absorbs any breaking change to the ApiClient shape, and the inner cast on Line 41 papers over the async-generator vs AsyncIterableStream mismatch. If subscribeToSessionStream ever gains required surface (or its return type changes), these tests will keep compiling while the manager breaks at runtime against the real client. Consider typing the partial mock as Pick<ApiClient, "subscribeToSessionStream"> and avoiding the outer double-cast — the manager only depends on that one method.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/sessionStreams/manager.test.ts` around lines 13 - 44,
The mock currently double-casts to ApiClient (singleShotApiClient) which hides
shape mismatches; change the mock's type to return Pick<ApiClient,
"subscribeToSessionStream"> instead of ApiClient and remove the outer `as
unknown as ApiClient` cast; ensure the subscribeToSessionStream method signature
and its returned async-iterable value match ApiClient's actual return type (use
ReturnType<ApiClient["subscribeToSessionStream"]> / Awaited<...> for the
function return) so you only need to satisfy the single method the manager uses
(subscribeToSessionStream) and avoid masking future API shape changes; keep
SSEStreamPart usage and the delivered logic intact.

packages/core/src/v3/realtimeStreams/streamsWriterV2.test.ts (1)

5-16: ⚡ Quick win

Assert lastEventId on the success path.

The new contract here is StreamWriteResult, but this test only proves that wait() resolves. If wait() regresses back to {} or undefined, this suite still passes. Seed lastAckedPosition() with a known seqNum and assert the returned { lastEventId }.

Also applies to: 132-148

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/realtimeStreams/streamsWriterV2.test.ts` around lines 5
- 16, The test currently stubs lastAckedPosition and appendSession but never
asserts the StreamWriteResult.lastEventId; update the mock for lastAckedPosition
(e.g., have lastAckedPosition return a known seqNum/object) and add an assertion
after calling wait() that the resolved value includes lastEventId equal to that
known seqNum; apply the same change to the second occurrence referenced (the
block around lines 132-148) so both success-path tests verify lastEventId from
lastAckedPosition via the appendSession mock and wait() result.

packages/core/src/v3/realtimeStreams/sessionStreamInstance.ts (2)

45-50: 💤 Low value

Best practice: Specify radix parameter for parseInt.

Lines 46 and 49 call parseInt without the radix parameter. While the header values should be base-10 strings, explicitly passing 10 as the radix prevents potential issues and follows best practices.

♻️ Add radix parameter

     const flushIntervalMs = headers["x-s2-flush-interval-ms"]
-      ? parseInt(headers["x-s2-flush-interval-ms"])
+      ? parseInt(headers["x-s2-flush-interval-ms"], 10)
       : undefined;
     const maxRetries = headers["x-s2-max-retries"]
-      ? parseInt(headers["x-s2-max-retries"])
+      ? parseInt(headers["x-s2-max-retries"], 10)
       : undefined;

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/realtimeStreams/sessionStreamInstance.ts` around lines
45 - 50, The parseInt calls used to parse headers["x-s2-flush-interval-ms"] and
headers["x-s2-max-retries"] (producing flushIntervalMs and maxRetries) should
specify a radix to avoid ambiguous parsing; update those parseInt invocations to
pass 10 as the second argument (e.g., parseInt(..., 10)) so the header strings
are parsed explicitly as base-10 integers.

---

29-38: 💤 Low value

Optional: Remove redundant optional chaining.

Line 37 uses this.options?.requestOptions, but options is a required constructor parameter and cannot be undefined. The ?. operator can be simplified to ..

♻️ Simplify optional chaining

     const response = await this.options.apiClient.initializeSessionStream(
       this.options.sessionId,
       this.options.io,
-      this.options?.requestOptions
+      this.options.requestOptions
     );

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/core/src/v3/realtimeStreams/sessionStreamInstance.ts` around lines
29 - 38, The code uses redundant optional chaining for requestOptions in
initializeWriter; since options is required on the SessionStreamInstance
constructor (type SessionStreamInstanceOptions<T>), replace
this.options?.requestOptions with this.options.requestOptions inside the
initializeWriter method (the call to
this.options.apiClient.initializeSessionStream) and remove the unnecessary ?. to
simplify the expression.

apps/webapp/app/presenters/v3/SessionPresenter.server.ts (1)

99-109: 💤 Low value

Optional: Consider extracting currentRun lookup for clarity.

The ternary and nullish coalescing on lines 99-108 are correct but require a non-null assertion on line 105. While this is safe (the outer ternary ensures session.currentRunId is truthy), extracting this logic into a helper or using an if-statement could improve readability.

♻️ Alternative: Extract to helper or use if-statement

-    const currentRun = session.currentRunId
-      ? runsById.get(session.currentRunId) ??
-        (await startActiveSpan(
-          "SessionPresenter.findCurrentRunFallback",
-          () =>
-            this.replica.taskRun.findFirst({
-              where: { id: session.currentRunId! },
-              select: { id: true, friendlyId: true, status: true },
-            })
-        ))
-      : null;
+    let currentRun = null;
+    if (session.currentRunId) {
+      currentRun = runsById.get(session.currentRunId);
+      if (!currentRun) {
+        currentRun = await startActiveSpan(
+          "SessionPresenter.findCurrentRunFallback",
+          () =>
+            this.replica.taskRun.findFirst({
+              where: { id: session.currentRunId },
+              select: { id: true, friendlyId: true, status: true },
+            })
+        );
+      }
+    }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/webapp/app/presenters/v3/SessionPresenter.server.ts` around lines 99 -
109, The currentRun lookup uses a nested ternary/nullish coalescing with a
non-null assertion on session.currentRunId; extract this into a small helper or
an if-statement to improve readability and remove the need for the assertion.
Specifically, replace the inline expression that builds currentRun (which
references session.currentRunId, runsById,
startActiveSpan("SessionPresenter.findCurrentRunFallback", ...), and
this.replica.taskRun.findFirst) with a clear helper function (e.g.,
findCurrentRun) or an explicit if-block that: 1) checks if session.currentRunId
is set, 2) returns runsById.get(session.currentRunId) if present, and 3)
otherwise calls startActiveSpan(...)/this.replica.taskRun.findFirst to fetch and
return the run; update the currentRun assignment to call that helper or use the
if-block so no non-null assertion is required.

apps/webapp/app/v3/services/createBackgroundWorker.server.ts (1)

348-348: ⚡ Quick win

The as any cast bypasses type safety.

Casting task.agentConfig as any removes compile-time guarantees that the config is JSON-serializable. While this may be necessary for Prisma's Json type, consider using a Zod schema to validate the config structure before persistence, or at minimum add a comment explaining why the cast is safe.

🛡️ Safer alternative with Zod validation

As per coding guidelines, use Zod for validation in apps/webapp:

+import { z } from "zod";
+
+const AgentConfigSchema = z.record(z.unknown()).optional();
+
 async function createWorkerTask(
   task: TaskResource,
   ...
 ) {
   ...
+  const validatedConfig = task.agentConfig 
+    ? AgentConfigSchema.parse(task.agentConfig) 
+    : undefined;
+
   await prisma.backgroundWorkerTask.create({
     data: {
       ...
-      config: task.agentConfig ? (task.agentConfig as any) : undefined,
+      config: validatedConfig as any,
     },
   });
 }

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@apps/webapp/app/v3/services/createBackgroundWorker.server.ts` at line 348,
The cast "task.agentConfig as any" in the createBackgroundWorker flow removes
type safety and should be replaced by runtime validation: define a Zod schema
(e.g., AgentConfigSchema) that matches the expected JSON-serializable shape,
call AgentConfigSchema.parse or safeParse on task.agentConfig inside the
function that builds the Prisma payload (the spot assigning config), and use the
validated value (or undefined) instead of the as any cast; if validation fails,
handle the error (reject/create a clear log and abort persisting) or default to
undefined. If a Zod schema cannot be introduced now, add a clear comment
explaining why the cast is safe and what invariants guarantee
JSON-serializability, referencing the config assignment (config:
task.agentConfig ...) so future reviewers know why it's exempt.

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes